package net.engio.mbassy.camel;

import net.engio.mbassy.MBassador;

import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The Mbassador consumer.
 */
public class MbassadorConsumer extends DefaultConsumer {

    /**
     * ID of event bus listener, registered by Camel.
     */
    private static final String LISTENER_ID = "CAMEL_LISTENER";

    private static final transient Logger LOG = LoggerFactory.getLogger(MbassadorConsumer.class);

    /**
     * MBassador event bus object instance.
     */
    private final MBassador eventBus;

    /**
     * Event bus listener object, registered by Camel.
     */
    private final CamelEventListener eventListener;

    /**
     * Creates a new instance of the MbassadorConsumer class.
     * @param endpoint {@link MbassadorEndpoint} object instance
     * @param processor {@link Processor} object instance
     * @param eventBus {@link MBassador} event bus object instance
     * @param eventClass {@link Class} of events received from the Mbassador to be filtered
     */
    public MbassadorConsumer(final MbassadorEndpoint endpoint, final Processor processor, final MBassador<?> eventBus,
                             final Class<?> eventClass) {
        super(endpoint, processor);
        this.eventBus = eventBus;
        this.eventListener = new CamelEventListener(LISTENER_ID, endpoint, processor, eventClass);
    }

    @Override
    protected void doStart() throws Exception {
        super.doStart();
        LOG.debug("Registering event listener: {} to MBassador: {}", eventListener, eventBus);
        eventBus.subscribe(eventListener);
    }

    @Override
    protected void doStop() throws Exception {
        LOG.debug("Unregistering event listener: {} from MBassador: {}", eventListener, eventBus);
        eventBus.unsubscribe(eventListener);
        super.doStop();
    }
}
